package com.ruoyi.kafka.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruoyi.kafka.domain.MessageEnvelope;
import com.ruoyi.kafka.domain.OrderCreatedEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @ClassName OrderConsumer
 * @Description 消费者
 * @Author li yang
 * @Date 2025/9/18 16:59
 * @Version 1.0
 */
@Slf4j
@Component
public class OrderConsumer {

    private final ObjectMapper objectMapper;

    public OrderConsumer(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    // 消费监听
    @KafkaListener(topics = "orders", groupId = "ly-service")
    public void onMessage(ConsumerRecord<String, MessageEnvelope<?>> record,
                          Acknowledgment ack) {
        try {
            MessageEnvelope<?> env = record.value();

            // 按 payloadClass 转回强类型（这里我们只演示 OrderCreatedEvent）
            Object payload = env.getPayload();
            Object typedPayload = payload;

            if (env.getPayloadClass() != null) {
                Class<?> clazz = Class.forName(env.getPayloadClass());
                typedPayload = objectMapper.convertValue(payload, clazz);
            }

            if (typedPayload instanceof OrderCreatedEvent) {
                OrderCreatedEvent event = (OrderCreatedEvent) typedPayload;

                log.info("收到消息: type={} id={} payload={} partition={} offset={}",
                        env.getType(), env.getId(), event, record.partition(), record.offset());

                // 示例：负数金额触发异常，走重试/最终进 DLT
                if (event.getAmount() != null && event.getAmount().signum() < 0) {
                    throw new RuntimeException("Invalid amount");
                }
            } else {
                log.info("收到未知类型 payload: class={}, value={}",
                        env.getPayloadClass(), typedPayload);
            }

            // 成功处理后手动提交位点
            ack.acknowledge();
        } catch (Exception e) {
            log.error("处理消息时发生错误: {}", e.getMessage());
            // 不提交位点，交给 DefaultErrorHandler 处理（重试->DLT）
            throw new RuntimeException(e);
        }
    }

}
